package com.wulaobo.wc;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @Author wuwenbo
 * @Date 2024/5/11 11:40
 * @ClassName: SocketStreamWordCount
 * @Description: TODO
 */
public class SocketStreamWordCount {

    public static void main(String[] args) throws Exception {

//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        env.setParallelism(1);
        env.disableOperatorChaining();

        DataStreamSource<String> lineStream = env.socketTextStream("111.229.26.142", 7777);

        SingleOutputStreamOperator<Tuple2<String, Integer>> streamOperator = lineStream.flatMap((String line, Collector<Tuple2<String, Integer>> collector) -> {
            String[] split = line.split(" ");
            for (String word : split) {
                if (StringUtils.isNotBlank(word)) {
                    Tuple2<String, Integer> tuple2 = Tuple2.of(word, 1);
                    collector.collect(tuple2);
                }
            }
        }).returns(Types.TUPLE(Types.STRING, Types.INT));
        KeyedStream<Tuple2<String, Integer>, String> keyedStream = streamOperator.keyBy(f -> f.f0);
        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = keyedStream.sum(1);
        sum.print();

        env.execute();

    }

}
